package com.lvmama.rhino.analyze.processing

import com.lvmama.rhino.common.entity.{DETAIL, LogType, PENDING_PAYMENT, WirelessClientMonitorLog}
import com.lvmama.rhino.common.utils.JDBCUtil.Conversions._
import com.lvmama.rhino.common.utils.Utils
import com.lvmama.rhino.common.utils.logging.LazyLogging
import com.lvmama.rhino.common.utils.methodset.InterfaceInvoke
import org.apache.spark.sql.Dataset
import org.apache.spark.sql.functions._
import org.apache.spark.storage.StorageLevel
import org.joda.time.{DateTime, Years}

/**
  * Created by yuanxiaofeng on 2017/5/19.
  */
class WirelessV2VisitorBatch extends Serializable with LazyLogging {
  self =>

  case class VisitorType(logType: String, pageTypeCode: String, stationId: String, platformCode: String,
                         timestamp: Long, channelCode: String, pageCode: String, province: String,
                         productId: String, sessionId: String, buttonCode: String, newVisitor: String,
                         specialSelling: String, productName: String, categoryId: String, deviceToken: String,
                         accountCreatedDate: String, paymentAmount: String, userId: String, userUseYears: String, ip: String)

  def process(ds: Dataset[WirelessClientMonitorLog]) = {

    val schemas = Utils.extractFieldNames[self.VisitorType].dropRight(1)
    org.apache.spark.sql.catalyst.encoders.OuterScopes.addOuterScope(self)

    import ds.sparkSession.implicits._
    val detail = ds
      .filter($"pageTypeCode" === DETAIL.code || $"pageTypeCode" === PENDING_PAYMENT.code)
      .filter($"logType" === LogType.forward.toString)
      .withColumn("productId", col("pageParam").getItem("productId"))
      .withColumn("specialSelling", col("pageParam").getItem("specialSelling"))
      .withColumn("productName", col("pageParam").getItem("productName"))
      .withColumn("categoryId", col("pageParam").getItem("categoryId"))
      .withColumn("paymentAmount", col("pageParam").getItem("paymentAmount"))
      .withColumn("accountCreatedDate", lit(""))
      .withColumn("userUseYears", lit(""))
      .select(schemas.map(col(_)): _*)
      .as[VisitorType]
      .persist(StorageLevel.MEMORY_AND_DISK_SER)

    val yesterday = Utils.getYesterday()
    import spray.json._
    import DefaultJsonProtocol._

    //PV
    detail.filter(d => d.pageTypeCode == DETAIL.code)
      .map(d => (d.platformCode,d.stationId, d.categoryId, if (d.specialSelling == "") "false" else d.specialSelling, d.productId, d.productName))
      .map(m => ((m._1, m._2, m._3, m._4), (m._5, m._6)))
      .groupByKey(g => g._1)
      .mapGroups { (key, iter) =>
        val _iter = iter.map(_._2).toList
        (key, _iter.size, _iter.groupBy(g => g).map(m => (m._1, m._2.size)))
      }
      .map(m => (m._1._1, m._1._2, m._1._3, m._1._4, m._2, Utils.emojiFilter(m._3.map(mm => (mm._1._1, mm._1._2, mm._2)).toJson.toString()), yesterday))
      .toDF("platform_code", "station_id", "category_id", "special_sell_flag", "pv", "product_json", "oper_date")
      .insertDF2MysqlDirect("flow_stat_product_details_pv")

    //UV IP
    detail.filter(d => d.pageTypeCode == DETAIL.code)
      .map(d => (d.ip, d.platformCode, d.productId, d.timestamp, d.stationId, d.categoryId, if (d.specialSelling == "") "false" else d.specialSelling, d.productName))
      .groupByKey(g => (g._1, g._2, g._3))
      .reduceGroups((x, y) => if (x._4 < y._4) x else y)
      .map(m => ((m._1._2, m._2._5, m._2._6, m._2._7), (m._2._3, m._2._8)))
      .groupByKey(g => g._1)
      .mapGroups { (key, iter) =>
        val _iter = iter.map(_._2).toList
        (key, _iter.size, _iter.groupBy(g => g).map(m => (m._1, m._2.size)))
      }
      .map(m => (m._1._1, m._1._2, m._1._3, m._1._4, m._2, Utils.emojiFilter(m._3.map(mm => (mm._1._1, mm._1._2, mm._2)).toJson.toString()), yesterday))
      .toDF("platform_code", "station_id", "category_id", "special_sell_flag", "uv", "product_json", "oper_date")
      .insertDF2MysqlDirect("flow_stat_product_details_uv")

    //Session 30分钟
    detail.filter(d => d.pageTypeCode == DETAIL.code)
      .map(d => (d.sessionId, d.platformCode, d.productId, d.timestamp, d.stationId, d.categoryId, if (d.specialSelling == "") "false" else d.specialSelling, d.productName))
      .groupByKey(g => (g._1, g._2, g._3))
      .reduceGroups((x, y) => if (x._4 < y._4) x else y)
      .map(m => ((m._1._2, m._2._5, m._2._6, m._2._7), (m._2._3, m._2._8)))
      .groupByKey(g => g._1)
      .mapGroups { (key, iter) =>
        val _iter = iter.map(_._2).toList
        (key, _iter.size, _iter.groupBy(g => g).map(m => (m._1, m._2.size)))
      }
      .map(m => (m._1._1, m._1._2, m._1._3, m._1._4, m._2, Utils.emojiFilter(m._3.map(mm => (mm._1._1, mm._1._2, mm._2)).toJson.toString()), yesterday))
      .toDF("platform_code", "station_id", "category_id", "special_sell_flag", "session", "product_json", "oper_date")
      .insertDF2MysqlDirect("flow_stat_product_details_session")


    //订单分析
    val now = DateTime.now()
    val orderNumAnalyze = detail
      .filter(d => d.pageTypeCode == PENDING_PAYMENT.code)
      //      .map(n => ((n.platformCode, n.categoryId, n.accountCreatedDate), (n.channelCode, n.paymentAmount)))
      //      .groupByKey(g => g._1)
      //      .mapGroups { (key, iter) =>
      //        val _iter = iter.map(_._2).toList
      //        (key, _iter.size, _iter.map(_._2).sum, _iter.groupBy(g => g._1).map(m => (m._1, m._2.size)))
      //      }
      .transform(rdd => {
      rdd.map { r =>
        val createTime = InterfaceInvoke.getUserCreatedDate(r.userId).split(" ")(0)
        val jodaTime = new DateTime(createTime)
        val delta = Years.yearsBetween(jodaTime, now).getYears
        r.copy(userUseYears = delta.toString)
      }
    })
      .groupByKey(g => (g.userUseYears, g.categoryId))
      .mapGroups { (key, iter) =>
        val _iter = iter.map(i => (i.channelCode, i.paymentAmount)).toList
        (key, _iter.size, _iter.map(_._2.toLong).sum, _iter.groupBy(g => g._1).map(m => (m._1, m._2.size)))
      }
      .map(m => (m._1._1, m._1._2, m._2, m._3, m._4.toJson.toString(), yesterday))
      .toDF("user_use_years", "category_id", "order_amount", "order_money", "order_json", "oper_date")

    //    orderNumAnalyze.insertDF2MysqlDirect("flow_stat_order")

    detail.unpersist()
  }
}
